/*******************************************************************************
 * Copyright 2013 Mindaugas Idzelis
 * 
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 * 
 *   http://www.apache.org/licenses/LICENSE-2.0
 * 
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 ******************************************************************************/
package incontrol.plugin.protocol;

import static com.google.common.base.Preconditions.checkNotNull;
import static com.google.common.collect.Lists.newArrayList;
import static com.google.common.collect.Maps.newHashMap;
import incontrol.core.protocol.IEvent;
import incontrol.core.protocol.IMessage;
import incontrol.core.protocol.IPluginRequest;
import incontrol.core.protocol.IResponse;
import incontrol.core.protocol.IServerRequest;
import incontrol.core.protocol.requests.RegisterPluginExtension;
import incontrol.core.protocol.responses.GenericResponse;

import java.lang.reflect.Method;
import java.util.Collection;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

import org.cometd.bayeux.Message;
import org.cometd.bayeux.client.ClientSessionChannel;
import org.cometd.bayeux.client.ClientSessionChannel.MessageListener;
import org.cometd.bayeux.server.LocalSession;

import com.google.common.base.Throwables;
import com.google.common.cache.Cache;
import com.google.common.cache.CacheBuilder;
import com.google.common.cache.CacheLoader;
import com.google.common.cache.LoadingCache;
import com.google.common.cache.RemovalListener;
import com.google.common.cache.RemovalNotification;
import com.google.common.collect.HashMultimap;
import com.google.common.collect.ImmutableList;
import com.google.common.collect.Multimap;
import com.google.common.reflect.TypeToken;
import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.SettableFuture;
import com.google.common.util.concurrent.UncheckedExecutionException;

public class EventBus {
	private Cache<Integer, SettableFuture<GenericResponse>> pendingResponses = CacheBuilder
			.newBuilder()
			.expireAfterWrite(30, TimeUnit.SECONDS)
			.removalListener(
					new RemovalListener<Integer, SettableFuture<GenericResponse>>() {
						public void onRemoval(
								RemovalNotification<Integer, SettableFuture<GenericResponse>> notification) {

						};
					}).build();
	private Cache<Integer, List<SettableFuture<GenericResponse>>> pendingPluginResponses = CacheBuilder
			.newBuilder()
			.expireAfterWrite(30, TimeUnit.SECONDS)
			.removalListener(
					new RemovalListener<Integer, List<SettableFuture<GenericResponse>>>() {
						public void onRemoval(
								RemovalNotification<Integer, List<SettableFuture<GenericResponse>>> notification) {

						};
					}).build();

	private class BayeuxBridge implements MessageListener {
		@Override
		public void onMessage(ClientSessionChannel from, Message message) {
			IMessage o = (IMessage) message.getData();
			SettableFuture<GenericResponse> response = pendingResponses
					.getIfPresent(o.getId());
			if (response != null) {
				if (o.getData() instanceof GenericResponse) {
					response.set((GenericResponse) o.getData());
					pendingResponses.invalidate(o.getId());
				}
			}
			List<SettableFuture<GenericResponse>> pluginResponse = pendingPluginResponses
					.getIfPresent(o.getId());
			if (pluginResponse != null) {
				if (o.getData() instanceof GenericResponse) {
					int set = 0;
					for (SettableFuture<GenericResponse> settableFuture : pluginResponse) {
						set++;
						if (settableFuture.set((GenericResponse) o.getData())) {
							break;
						}
					}
					if (set == pluginResponse.size()) {
						pendingPluginResponses.invalidate(o.getId());
					}
				}
			}

		}
	}

	public EventBus(LocalSession session) {
		session.handshake();
		newLocalSession = session;

	}

	private static final LoadingCache<Class<?>, ImmutableList<Method>> handlerMethodsCache = CacheBuilder
			.newBuilder().weakKeys()
			.build(new CacheLoader<Class<?>, ImmutableList<Method>>() {
				@Override
				public ImmutableList<Method> load(Class<?> concreteClass)
						throws Exception {
					return getAnnotatedMethodsInternal(concreteClass);
				}
			});

	private static ImmutableList<Method> getAnnotatedMethodsInternal(
			Class<?> clazz) {
		Set<? extends Class<?>> supers = TypeToken.of(clazz).getTypes()
				.rawTypes();
		ImmutableList.Builder<Method> result = ImmutableList.builder();
		for (Method method : clazz.getMethods()) {
			/*
			 * Iterate over each distinct method of {@code clazz}, checking if
			 * it is annotated with
			 * 
			 * @Subscribe by any of the superclasses or superinterfaces that
			 * declare it.
			 */
			for (Class<?> c : supers) {
				try {
					Method m = c.getMethod(method.getName(),
							method.getParameterTypes());
					if (m.isAnnotationPresent(INSubscribe.class)) {
						Class<?>[] parameterTypes = method.getParameterTypes();
						if (parameterTypes.length != 1) {
							throw new IllegalArgumentException(
									"Method "
											+ method
											+ " has @Subscribe annotation, but requires "
											+ parameterTypes.length
											+ " arguments.  Event handler methods must require a single argument.");
						}
						Class<?> eventType = parameterTypes[0];
						result.add(method);
						break;
					}
				} catch (NoSuchMethodException ignored) {
					// Move on.
				}
			}
		}
		return result.build();
	}

	private Multimap<String, Method> channelToHandler = HashMultimap.create();
	private LocalSession newLocalSession;
	private Multimap<Object, ChannelListener> objectToListener = HashMultimap
			.create();
	private AtomicInteger currentId = new AtomicInteger();
	private Map<String, MessageListener> responseListener = newHashMap();

	public EventBus() {

	}

	public Multimap<String, Method> findAllHandlers(Object listener) {
		Multimap<String, Method> methodsInListener = HashMultimap.create();
		Class<?> clazz = listener.getClass();
		for (Method method : getAnnotatedMethods(clazz)) {
			Class<?>[] parameterTypes = method.getParameterTypes();
			Class<?> eventType = parameterTypes[0];
			Channel annotation = eventType.getAnnotation(Channel.class);
			if (annotation != null) {
				methodsInListener.put(annotation.name(), method);
			}
		}
		return methodsInListener;
	}

	private static ImmutableList<Method> getAnnotatedMethods(Class<?> clazz) {
		try {
			return handlerMethodsCache.getUnchecked(clazz);
		} catch (UncheckedExecutionException e) {
			throw Throwables.propagate(e.getCause());
		}
	}

	public void sendServerResponse(IPluginRequest request, IResponse response) {
		IMessage msg = activeRequests.get(request);
		Channel annotation = request.getClass().getAnnotation(Channel.class);
		if (annotation == null) {
			checkNotNull(request, "Posted object must have channel annotation");
		}
		String channel = annotation.name();

		IMessage wrapper = new IMessage(response, msg.getId(), null);
		newLocalSession.getChannel(channel).publish(wrapper);
	}

	public Future<GenericResponse> sendServerRequest(IServerRequest message) {
		Channel annotation = message.getClass().getAnnotation(Channel.class);
		if (annotation == null) {
			checkNotNull(message, "Posted object must have channel annotation");
		}
		String channel = annotation.name();

		IMessage wrapper = new IMessage(message, currentId.incrementAndGet(),
				null);
		wrapper.setData(message);

		SettableFuture<GenericResponse> future = SettableFuture
				.<GenericResponse> create();
		pendingResponses.put(wrapper.getId(), future);

		MessageListener listener = responseListener.get(channel);
		if (listener == null) {
			listener = new BayeuxBridge();
			responseListener.put(channel, listener);
			newLocalSession.getChannel(channel).subscribe(listener);
		}
		newLocalSession.getChannel(channel).publish(wrapper);
		return future;
	}

	public Future<List<GenericResponse>> sendPluginRequest(
			IPluginRequest pluginRequest, int responses) {
		Channel annotation = pluginRequest.getClass().getAnnotation(
				Channel.class);
		if (annotation == null) {
			checkNotNull(pluginRequest,
					"Posted object must have channel annotation");
		}
		String channel = annotation.name();

		IMessage wrapper = new IMessage(pluginRequest,
				currentId.incrementAndGet(), null);
		wrapper.setData(pluginRequest);

		List<SettableFuture<GenericResponse>> futures = newArrayList();
		for (int i = 0; i < responses; i++) {
			SettableFuture<GenericResponse> future = SettableFuture
					.<GenericResponse> create();
			futures.add(future);
		}

		pendingPluginResponses.put(wrapper.getId(), futures);

		MessageListener listener = responseListener.get(channel);
		if (listener == null) {
			listener = new BayeuxBridge();
			responseListener.put(channel, listener);
			newLocalSession.getChannel(channel).subscribe(listener);
		}
		newLocalSession.getChannel(channel).publish(wrapper);
		// returns a list of future responses, once they are all set.
		return Futures.allAsList(futures);

	}

	private Map<IPluginRequest, IMessage> activeRequests = newHashMap();

	public void register(final Object object) {
		Multimap<String, Method> findAllHandlers = findAllHandlers(object);
		for (String channel : findAllHandlers.keySet()) {
			Collection<Method> methods = findAllHandlers.get(channel);
			for (final Method method : methods) {

				MessageListener listener = new MessageListener() {

					@Override
					public void onMessage(ClientSessionChannel from,
							Message message) {

						Object o = message.getData();
						if (o instanceof IMessage) {
							o = ((IMessage) o).getData();
						}
						if (o instanceof IEvent) {
							IEvent msg = (IEvent) o;

							try {
								method.invoke(object, msg);
							} catch (Exception e) {
								e.printStackTrace();
							}
						} else if (o instanceof IPluginRequest) {
							IPluginRequest req = ((IPluginRequest) o);
							try {
								activeRequests.put(req,
										(IMessage) message.getData());
								method.invoke(object, req);
							} catch (Exception e) {
								e.printStackTrace();
							} finally {
								activeRequests.remove(req);
							}
						}

					}

				};

				objectToListener.put(object, new ChannelListener(channel,
						listener));
				newLocalSession.getChannel(channel).subscribe(listener);
				if (IPluginRequest.class.isAssignableFrom(method
						.getParameterTypes()[0])) {
					sendServerRequest(new RegisterPluginExtension(channel));
				}
			}
		}
	}

	private class ChannelListener {
		private String channel;
		private MessageListener listener;

		public ChannelListener(String channel, MessageListener listener) {
			super();
			this.channel = channel;
			this.listener = listener;
		}

	}

	public void unregister(Object object) {

		Collection<ChannelListener> removeAll = objectToListener
				.removeAll(object);
		for (ChannelListener cl : removeAll) {
			newLocalSession.getChannel(cl.channel).unsubscribe(cl.listener);
		}

	}

}
